package app.dwd;


import common.Constant;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import util.SQLUtil;

public class Dwd_Exam_Test extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_Exam_Test().init(
                2001,
                2,
                "Dwd_Exam_Test"
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env,
                          StreamTableEnvironment tEnv) {
        readOdsDb(tEnv,"Dwd_Exam_Test");



//        readBaseDic(tEnv);
//        试卷+课程+题目

        //一次考试
        Table testExam = tEnv.sqlQuery("select " +
                "       data['id'] id, " +
                "       data['paper_id'] paper_id, " +
                "       data['user_id'] user_id, " +
                "       data['score'] score, " +
                "       data['duration_sec'] duration_sec," +
                "       ts" +
                "       from ods_db " +
                "       where `database`='edu' " +
                "       and `table`='test_exam' " +
                "       and `type`='insert' " +
                "       and `data`['deleted']='0'");
        tEnv.createTemporaryView("testExam",testExam);
//        CloseableIterator<Row> collect = testExam.execute().collect();
//        System.out.println("=======test1==========");
//        while (collect.hasNext()){
//            System.out.println("=======test2==========");
//            System.out.println(collect.next());
//        }
//        DataStream<Tuple2<Boolean, Row>> tuple2DataStream = tEnv.toRetractStream(testExam, Row.class);
//        tuple2DataStream.print();


        //一份试卷
        Table examPaper = tEnv.sqlQuery("select " +
                "   data['id'] id,  " +
                "   data['paper_title'] paper_title, " +
                "   data['course_id'] course_id " +
                "   from ods_db " +
                "   where `database`='edu' " +
                "   and `table`='test_paper' " +
                "   and `type`='insert' " +
                "   and `data`['deleted']='0'");
        tEnv.createTemporaryView("courseInfo",examPaper);


        //一次答题记录
        Table oneQuestion = tEnv.sqlQuery("select " +
                "       data['id'] id, " +
                "       data['exam_id'] exam_id, " +
                "       data['paper_id'] paper_id, " +
                "       data['question_id'] question_id, " +
                "       data['user_id'] user_id, " +
                "       data['answer'] answer, " +
                "       data['is_correct'] is_correct, " +
                "       data['score'] score " +
                "from ods_db " +
                "where `database`='edu' " +
                "and `table`='test_exam_question' " +
                "and `type`='insert' " +
                "and `data`['deleted']='0'");
        tEnv.createTemporaryView("oneQuestion",oneQuestion);
//        CloseableIterator<Row> collect = oneQuestion.execute().collect();
//        System.out.println("=======test1==========");
//        while (collect.hasNext()){
//            System.out.println("=======test2==========");
//            System.out.println(collect.next());
//        }
//        DataStream<Tuple2<Boolean, Row>> tuple2DataStream = tEnv.toRetractStream(testExam, Row.class);
//        tuple2DataStream.print();

        //测验和试卷join可以拿出课程id
        Table result = tEnv.sqlQuery("select " +
                " te.id exam_id, " +
                " te.paper_id, " +
                " te.user_id, " +
                " ci.course_id, " +
                " ci.paper_title, " +
                " OQ.question_id, " +
                " te.score, " +
                " te.duration_sec, " +
                " OQ.answer," +
                " OQ.is_correct, " +
                " te.ts," +
                " current_row_timestamp() row_op_ts " +
                "from testExam te " +
                "left join courseInfo ci " +
                "on te.paper_id=ci.id " +
                "left join oneQuestion OQ " +
                "on te.user_id=OQ.user_id and OQ.paper_id=te.paper_id ");

        tEnv.executeSql("create table dwd_exam_test( " +
                "exam_id  string,   " +
                "paper_id string,   " +
                "user_id string,   " +
                "course_id string,   " +
                "paper_title string,   " +
                "question_id string,   " +
                "score string,   " +
                "duration_sec string,   " +
                "answer string,  " +
                "is_correct string,   " +
                "ts bigint,  " +
                "row_op_ts timestamp_ltz(3), " +
                "primary key(exam_id) not enforced" +
                ")" + SQLUtil.getUpsertKafkaSink(Constant.TOPIC_DWD_EXAM_TEST));
        result.executeInsert("dwd_exam_test");


    }
}